/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include "ApsaraStackProtectEngine.h"

using namespace OpenStackPlugin;

namespace {
    const std::string MODULE = "ApsaraStackProtectEngine";
    const std::string APSARASTACK_API_CLASS = "ResourceApiWrapper";
    const std::string COMMON_FORMAT = "ssssssisss";
    const std::string COMMON_RESOURCEGROUP = "RESOURCE_SET";
    const int32_t DEFAULT_TIMEOUT = 3000;
    const int32_t COMMON_RETRY_TIME = 5;
    const int32_t COMMON_WAIT_TIME = 10;
    const uint32_t GB_SIZE = 1024 * 1024 * 1024UL;
    const std::string APSARA_STACK_CONF = "ApsaraStackConfig";
    const std::string APSARA_STATUS_RUNNING = "Running";
    const std::string APSARA_STATUS_STOPPED = "Stopped";
    const int NOT_DELETE_SNAPSHOT_VOLUME = 0;
    const int MAX_DISK_NUMBER_IN_GROUP = 16;
    const int MAX_DISK_SIZE_IN_GROUP = 32 * 1024;
    const int DEFAULT_PAGE_NUMBER = 1;
    const int DEFAULT_PAGE_SIZE = 100;
    const std::string SNAPSHOT_QUOTA_EXCEED = "1577213595";    // 快照达到上限
};

namespace ApsaraStackPlugin {

void ApsaraStackProtectEngine::DiscoverApplications(std::vector<Application>& returnValue, const std::string& appType)
{
    return;
}

void ApsaraStackProtectEngine::CheckApplication(ActionResult &returnValue, const ApplicationEnvironment &appEnv,
    const AppProtect::Application &application)
{
    ApsaraStackResourceAccess resourceAccess(appEnv);
    resourceAccess.SetApplication(application);
    m_appEnv = appEnv;
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    if (resourceAccess.CheckAppConnect(returnValue) != SUCCESS) {
        ERRLOG("Failed to check application.");
        return;
    }
    INFOLOG("Success to check application.");
    return;
}

void ApsaraStackProtectEngine::ListApplicationResourceV2(ResourceResultByPage& page, const ListResourceRequest& request)
{
    ListResourcePara conditionPara;
    if (!Module::JsonHelper::JsonStringToStruct(request.condition.conditions, conditionPara)) {
        ERRLOG("Failed to read condition.");
        return;
    }
    m_appEnv = request.appEnv;
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    ApsaraStackResourceAccess apsaraStackResAccess(request.appEnv, request.condition);
    if (apsaraStackResAccess.ListResource(page, conditionPara) != SUCCESS) {
        ERRLOG("Failed to scan resource %s.", conditionPara.mResourceType.c_str());
        return;
    }
    return;
}

int32_t ApsaraStackProtectEngine::GetOrganizationInfo(const std::string &organizationId, Organization &org)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("s", organizationId.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "des_organization", pArgs,
        response) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return FAILED;
    }
    DesOrganizationResponse resourceResponse;
    if (Utils::TransStrToApsaraApiResponse(response, resourceResponse) != SUCCESS) {
        ERRLOG("Failed to parse list disks function result.");
        return FAILED;
    }
    if (std::to_string(resourceResponse.mData.mId) != organizationId) {
        ERRLOG("Describe organization failed, id: %d.", resourceResponse.mData.mId);
        return FAILED;
    }
    org = resourceResponse.mData;
    INFOLOG("Describe organization success.");
    return SUCCESS;
}

void ApsaraStackProtectEngine::DiscoverAppCluster(ApplicationEnvironment& returnEnv,
    const ApplicationEnvironment& appEnv, const Application& application)
{
    m_appEnv = appEnv;
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    ActionResult returnValue;
    Organization org;
    if (GetOrganizationInfo(m_envParam.m_organizationId, org) != SUCCESS) {
        ERRLOG("Failed to GetOrganizationInfo.");
        return;
    }
    INFOLOG("Success to DiscoverAppCluster.");
    Json::Value extendInfo;
    Json::FastWriter writer;
    extendInfo["clusterFlag"] = "0";
    returnEnv.__set_extendInfo(writer.write(extendInfo));
    return;
}

// snapshot
bool ApsaraStackProtectEngine::CheckIsConsistenSnapshot()
{
    std::string ifCreateConsistenSnapshot = Module::ConfigReader::getString("ApsaraStackConfig",
        "CreateConsistenSnapshot");
    return ifCreateConsistenSnapshot == "true";
}

bool ApsaraStackProtectEngine::DoCreateSnapshot(const std::vector<VolInfo> &volList,
    SnapshotInfo &snapshot, std::string &errCode)
{
    if (CheckIsConsistenSnapshot() && CheckIfCreateSnapGroup(volList)) {
        INFOLOG("Start create consistent snapshot, %s.", m_taskId.c_str());
        bool ret = CreateConsistencySnapshot(volList, snapshot, errCode);
        TP_START("TP_CreateSnapGroup", 1, &ret);
        TP_END
        return ret;
    }
    INFOLOG("Start create common snapshot, %s.", m_taskId.c_str());
    return DoCreateCommonSnapshot(volList, snapshot, errCode);
}

int32_t ApsaraStackProtectEngine::DeleteSnapshot(const SnapshotInfo &snapshot)
{
    if (snapshot.m_volSnapList.empty()) {
        DBGLOG("Snapshot list is empty, no volume snapshot to be deleted, %s", m_taskId.c_str());
        return SUCCESS;
    }
    if (CheckIsConsistenSnapshot()) {
        INFOLOG("Start Delete consistent snapshot, %s.", m_taskId.c_str());
        return DoDeleteConsistencySnapshot(snapshot);
    }

    INFOLOG("Start Delete common snapshot, %s.", m_taskId.c_str());
    return DeleteCommonSnapshot(snapshot);
}

bool ApsaraStackProtectEngine::CheckIfCreateSnapGroup(const std::vector<VolInfo> &volList)
{
    int32_t totalSize = 0;
    if (volList.size() > MAX_DISK_NUMBER_IN_GROUP) {
        WARNLOG("Disk number is more than 16, cant create snapshot group.");
        return false;
    }
    for (const auto& disk : volList) {
        if (disk.m_volumeType != "cloud_pperf" && disk.m_volumeType != "cloud_sperf") {
            WARNLOG("Disk type %s dont support create snapshot group.", disk.m_volumeType.c_str());
            return false;
        }
        totalSize += disk.m_volSizeInBytes / GB_SIZE;
    }
    if (totalSize > MAX_DISK_SIZE_IN_GROUP) {
        WARNLOG("Disk total size is more than 32TB, cant create snapshot group.");
        return false;
    }
    return true;
}

bool ApsaraStackProtectEngine::CreateConsistencySnapshot(const std::vector<VolInfo> &volList,
    SnapshotInfo &snapshot, std::string &errCode)
{
    std::string snapGroupName = "SnapshotGroupOfOP_" + m_requestId;
    std::vector<std::string> excludeIdList;
    for (const auto &vol : volList) {
        if (m_volMap.find(vol.m_uuid) == m_volMap.end()) {
            WARNLOG("Don't backup vol %s.", vol.m_uuid.c_str());
            excludeIdList.push_back(vol.m_uuid);
        }
    }
    std::string excludeIdStr = boost::algorithm::join(excludeIdList, ",");
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ssss", m_instance.m_regionId.c_str(), m_instance.m_id.c_str(),
            snapGroupName.c_str(), excludeIdStr.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS,
        "create_snapshotgroup", pArgs, response) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return false;
    }
    CreateSnapshotGroupResponse createRes;
    if (Utils::TransStrToApsaraApiResponse(response, createRes) != SUCCESS) {
        ERRLOG("Failed to parse create snapshot group function result.");
        return false;
    }
    if (createRes.m_code == "QuotaExceed.Snapshot") {
        errCode = SNAPSHOT_QUOTA_EXCEED;
        ERRLOG("Snapshot quota exceed.");
        return false;
    }
    if (!WaitSnapshotGroup(createRes, snapGroupName, snapshot)) {
        ERRLOG("Wait snapshot group(%s) accomplished timeout.", createRes.m_snapshotGroupId.c_str());
        return false;
    }
    INFOLOG("Create snapshot group %s success.", createRes.m_snapshotGroupId.c_str());
    return true;
}

bool ApsaraStackProtectEngine::WaitSnapshotGroup(const CreateSnapshotGroupResponse &createRes,
    const std::string & snapGroupName, SnapshotInfo &snapshot)
{
    APSSnapshotGroup snapshotGroup;
    int retryTimes = Module::ConfigReader::getInt("ApsaraStackConfig", "CreateSnapshotWaitRetryTimes");
    int waitTime =  Module::ConfigReader::getInt("ApsaraStackConfig", "CreateSnapshotWaitInterval");
    retryTimes = retryTimes > 0 ? retryTimes : COMMON_RETRY_TIME;
    waitTime = waitTime > 0 ? waitTime : COMMON_WAIT_TIME;
    for (int i = 0; i < retryTimes; ++i) {
        if (GetSnapshotGroupDetail(createRes.m_snapshotGroupId, snapshotGroup, snapshot) != SUCCESS) {
            ERRLOG("Call python funtion failed.");
            return false;
        }
        if (snapshotGroup.m_status == "failed") {
            ERRLOG("Snapshot group %s status is erro.", snapGroupName.c_str());
            return false;
        }
        if (snapshotGroup.m_status == "accomplished") {
            INFOLOG("Create snapshot group %s success.", snapGroupName.c_str());
            return true;
        }
        WARNLOG("Snapshot group status is %s.", snapshotGroup.m_status.c_str());
        sleep(waitTime);
    }
    ERRLOG("Create snapshot group %s time out.", snapGroupName.c_str());
    return false;
}

int32_t ApsaraStackProtectEngine::GetSnapshotGroupDetail(const std::string &groupId, APSSnapshotGroup &snapshotGroup,
    SnapshotInfo &snapshot)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ss", m_regionId.c_str(), groupId.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_snapshotgroup",
        pArgs, response)
        != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return FAILED;
    }
    APSGetSnapshotGroupResponse snapshotGroups;
    if (Utils::TransStrToApsaraApiResponse(response, snapshotGroups) != SUCCESS) {
        ERRLOG("Failed to parse get snapshot group function result.");
        return FAILED;
    }
    if (snapshotGroups.m_snapshotGroups.m_snapshotGroup.size() != 1) {
        ERRLOG("Failed to get snapshot group %s detail.", groupId.c_str());
        return FAILED;
    }
    snapshotGroup = snapshotGroups.m_snapshotGroups.m_snapshotGroup[0];
    snapshot.m_consistengroupId = snapshotGroup.m_snapshotGroupId;
    snapshot.m_isconsistentSnapshot = true;
    snapshot.m_vmMoRef = snapshotGroup.m_instanceId;
    for (const auto &diskSnap : snapshotGroup.m_snapshots.m_snapshot) {
        VolSnapInfo volSnap;
        volSnap.m_volUuid = diskSnap.m_sourceDiskId;
        volSnap.m_snapshotId = diskSnap.m_id;
        volSnap.m_snapshotName = diskSnap.m_name;
        volSnap.m_status = diskSnap.m_status;
        snapshot.m_volSnapList.push_back(volSnap);
    }
    DBGLOG("Get snapshot group %s detail success.", groupId.c_str());
    return SUCCESS;
}

bool ApsaraStackProtectEngine::DoCreateCommonSnapshot(const std::vector<VolInfo> &volList,
    SnapshotInfo &snapshot, std::string &errCode)
{
    DBGLOG("Start to create snapshots. %s", m_taskId.c_str());
    int32_t ret = SUCCESS;
    TP_START("TP__CreateSnap", 1, &ret);
    TP_END
    if (ret == FAILED) {
        return false;
    }
    int32_t remainVolListSize = volList.size();
    for (const auto &vol : volList) {
        VolSnapInfo volSnap;
        if (!CreateDiskSnapshot(vol, volSnap, errCode)) {
            ERRLOG("Create disk %s snapshot failed.", vol.m_uuid.c_str());
            return false;
        }
        snapshot.m_volSnapList.push_back(volSnap);
    }
    return true;
}

bool ApsaraStackProtectEngine::CreateDiskSnapshot(const VolInfo &volInfo, VolSnapInfo &volSnap, std::string &errCode)
{
    std::string snapName = GenerateSnapshotName(volInfo.m_uuid);
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("sss", volInfo.m_uuid.c_str(), snapName.c_str(), m_snapDescription.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "create_snapshot",
        pArgs, response) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return false;
    }
    APSCreateSnapshotResponse snapshotRes;
    if (Utils::TransStrToApsaraApiResponse(response, snapshotRes) != SUCCESS) {
        ERRLOG("Failed to parse create snapshot group function result.");
        return false;
    }
    APSSnapshot snapshot;
    int32_t err;
    for (int i = 0; i < COMMON_RETRY_TIME; ++i) {
        if (GetSnapshotDetail(snapshotRes.m_snapshotId, snapshot, err) != SUCCESS) {
            ERRLOG("Get snapshot detail failed.");
            return false;
        }
        volSnap.m_snapshotId = snapshot.m_id;
        if (snapshot.m_status == "failed") {
            ERRLOG("Snapshot %s status is erro.", snapName.c_str());
            if (!DeleteVolumeSnapshot(volSnap)) {
                ERRLOG("Delete disk snapshot failed, volId: %s, snapId: %s, taskId: %s", volInfo.m_uuid.c_str(),
                    volSnap.m_snapshotId.c_str(), m_taskId.c_str());
            }
            return false;
        }
        if (snapshot.m_status == "accomplished") {
            INFOLOG("Create snapshot  %s success.", snapName.c_str());
            FillUpVolSnapInfo(volInfo, snapshot, volSnap);
            return true;
        }
        sleep(COMMON_WAIT_TIME);
    }
    ERRLOG("Create snapshot %s time out.", snapName.c_str());
    return false;
}

bool ApsaraStackProtectEngine::SendDeleteSnapshotMsg(const VolSnapInfo &volSnap)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("s", volSnap.m_snapshotId.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "delete_snapshot",
        pArgs, response) != SUCCESS) {
        ERRLOG("Call python funtion delete_snapshot failed.");
        return false;
    }
    int32_t ret = SUCCESS;
    TP_START("TP_DeleteSnapShot", 1, &ret);
    TP_END
    if (ret == FAILED) {
        return false;
    }
    INFOLOG("Send delete snapshot %s request.", volSnap.m_snapshotId.c_str());
    return true;
}

bool ApsaraStackProtectEngine::ConfirmIfSnapDeleted(const std::string &snapId, int retryTimes,
    int &curConsumeRetryTimes)
{
    APSSnapshot snapshot;
    int32_t erro;
    uint32_t retry = 0;
    while (retry++ < retryTimes) {
        curConsumeRetryTimes++;
        if (GetSnapshotDetail(snapId, snapshot, erro) == SUCCESS) {
            WARNLOG("Delete snapshot %s failed, snapshot status: %s.", snapId.c_str(),
                snapshot.m_status.c_str());
            sleep(COMMON_WAIT_TIME);
        }
        if (erro == Module::SC_NOT_FOUND) {
            INFOLOG("Delete snapshot %s success.", snapId.c_str());
            return true;
        }
    }
    return false;
}

void ApsaraStackProtectEngine::FillUpVolSnapInfo(const VolInfo &volInfo, const APSSnapshot &snapDetails,
    VolSnapInfo &volSnap)
{
    volSnap.m_volUuid = snapDetails.m_sourceDiskId;
    volSnap.m_snapshotName = snapDetails.m_name;
    volSnap.m_status = snapDetails.m_status;
    volSnap.m_extendInfo = volInfo.m_volumeType;
    volSnap.m_snapshotId = snapDetails.m_id;
    return;
}

int32_t ApsaraStackProtectEngine::GetSnapshotDetail(const std::string snapshotId, APSSnapshot &snapshot,
    int32_t &err)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ss", m_regionId.c_str(), snapshotId.c_str());
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_snapshot", pArgs,
        rspStr) != SUCCESS) {
        ERRLOG("Call python funtion list_snapshot failed.");
        return FAILED;
    }
    APSGetSnapshotResponse response;
    if (Utils::TransStrToApsaraApiResponse(rspStr, response) != SUCCESS) {
        ERRLOG("Failed to parse get snapshot function result.");
        return FAILED;
    }
    if (response.m_snapshots.m_snapshot.size() == 0) {
        ERRLOG("Not find snapshot %s.", snapshotId.c_str());
        err = Module::SC_NOT_FOUND;
        return FAILED;
    }
    snapshot = response.m_snapshots.m_snapshot[0];
    INFOLOG("Get snapshot %s detail success.", snapshotId.c_str());
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::DoDeleteConsistencySnapshot(const SnapshotInfo &snapshot)
{
    int32_t ret = SUCCESS;
    TP_START("TP_DeleteSnapShot", 1, &ret);
    TP_END
    if (ret == FAILED) {
        return FAILED;
    }
    std::set<std::string> snapshotGroupIds;
    for (const VolSnapInfo &volSnap : snapshot.m_volSnapList) {
        if (volSnap.m_snapshotName.find("Created from ") != std::string::npos) {
            int32_t pos = volSnap.m_snapshotName.find("ssg-");
            std::string snapshotGroupId = volSnap.m_snapshotName.substr(pos);
            DBGLOG("Snapshot group id %s.", snapshotGroupId.c_str());
            snapshotGroupIds.insert(snapshotGroupId);
        }
    }
    for (const auto &groupId : snapshotGroupIds) {
        if (DeleteSnapshotGroup(groupId) != SUCCESS) {
            ERRLOG("Delete snapshot group %s failed.", groupId.c_str());
            ret = FAILED;
        }
    }
    return ret;
}

int32_t ApsaraStackProtectEngine::DeleteSnapshotGroup(const std::string &groupId)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ss", m_regionId.c_str(), groupId.c_str());
    }
    SnapshotInfo deleteFailedSnapshot;
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "delete_snapshotgroup",
        pArgs, rspStr)
        != SUCCESS) {
        ERRLOG("Call python funtion list_snapshot failed.");
        return FAILED;
    }
    APSDeleteSnapshotGroupResponse response;
    if (Utils::TransStrToApsaraApiResponse(rspStr, response) != SUCCESS) {
        ERRLOG("Failed to parse delete snapshot group function result.");
        return FAILED;
    }
    for (const auto &item : response.m_operationProgressSet.m_operationProgress) {
        if (item.m_operationStatus != "Success") {
            ERRLOG("Delete snapshot group %s failed, errMsg: %s.", groupId.c_str(),
                item.m_errorMsg.c_str());
            return FAILED;
        }
    }
    INFOLOG("Delete snapshot group %s success.", groupId.c_str());
    return SUCCESS;
}

bool ApsaraStackProtectEngine::GetVolumeSnapshot(VolSnapInfo &volSnap)
{
    int32_t erro;
    APSSnapshot snapshot;
    if (GetSnapshotDetail(volSnap.m_snapshotId, snapshot, erro) != SUCCESS) {
        if (erro == Module::SC_NOT_FOUND) {
            INFOLOG("Delete snapshot %s success.", volSnap.m_snapshotId.c_str());
            volSnap.m_deleted = true;
            return true;
        }
        INFOLOG("Get snapshot %s detail failed.", volSnap.m_snapshotId.c_str());
        return false;
    }
    INFOLOG("Get snapshot %s detail success, snapshot status: %s.", volSnap.m_snapshotId.c_str(),
        snapshot.m_status.c_str());
    volSnap.m_status = snapshot.m_status;
    return true;
}

int32_t ApsaraStackProtectEngine::GetSnapshotsOfVolume(const VolInfo &volInfo, std::vector<VolSnapInfo> &snapList)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("sss", m_regionId.c_str(), "", volInfo.m_uuid.c_str());
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_snapshot",
        pArgs, rspStr) != SUCCESS) {
        ERRLOG("Call python funtion list_snapshot failed.");
        return FAILED;
    }
    APSGetSnapshotResponse response;
    if (Utils::TransStrToApsaraApiResponse(rspStr, response) != SUCCESS) {
        ERRLOG("Failed to parse get snapshot function result.");
        return FAILED;
    }
    if (response.m_snapshots.m_snapshot.size() == 0) {
        WARNLOG("Not find snapshot of disk %s.", volInfo.m_uuid.c_str());
    }
    for (const APSSnapshot &snap : response.m_snapshots.m_snapshot) {
        if (MatchSnapshotOfBackup(snap) != SUCCESS) {
            DBGLOG("Snapshot name(%s) not match.", snap.m_name.c_str());
            continue;
        }
        VolSnapInfo volSnap;
        FillUpVolSnapInfo(volInfo, snap, volSnap);
        snapList.push_back(volSnap);
    }
    INFOLOG("Get snapshot of disk %s success.", volInfo.m_uuid.c_str());
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::MatchSnapshotOfBackup(const APSSnapshot &snap)
{
    static std::unordered_map<std::string, int32_t> snapshotGroups;
    // 一致性快照组中快照的名字，"Created from ssg-xxxxxxxxxxxxxxxx"
    if (snap.m_name.find("Created from ") != std::string::npos) {
        int32_t pos = snap.m_name.find("ssg-");
        std::string snapshotGroupId = snap.m_name.substr(pos);
        if (snapshotGroups.find(snapshotGroupId) != snapshotGroups.end()) {
            return snapshotGroups[snapshotGroupId];
        }
        APSSnapshotGroup snapshotGroup;
        SnapshotInfo snapshot;
        if (GetSnapshotGroupDetail(snapshotGroupId,  snapshotGroup, snapshot) != SUCCESS) {
            snapshotGroups[snapshotGroupId] = FAILED;
            return FAILED;
        }
        if (snapshotGroup.m_name.find("SnapshotGroupOfOP_") != std::string::npos) {
            snapshotGroups[snapshotGroupId] = SUCCESS;
            return SUCCESS;
        }
        snapshotGroups[snapshotGroupId] = FAILED;
        return FAILED;
    }
    // 创建的普通快照
    if (!MatchSnapshotName(snap.m_name) || !MatchSnapshotDescription(snap.m_description)) {
        ERRLOG("Snapshot name(%s) or description(%s) not match.", snap.m_name.c_str(), snap.m_description.c_str());
        return FAILED;
    }
    return SUCCESS;
}

int ApsaraStackProtectEngine::PreHook(const ExecHookParam &para)
{
    if (!InitJobPara()) {
        ERRLOG("Init Job Para failed.");
        return FAILED;
    }
    InstanceResource serverExtendInfo;
    if (!Module::JsonHelper::JsonStringToStruct(m_application.extendInfo, serverExtendInfo)) {
        ERRLOG("FormHttpRequest failed, %s", m_taskId.c_str());
        return FAILED;
    }
    m_regionId = serverExtendInfo.mRegionId;
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    if (m_jobHandle->GetJobType() == JobType::BACKUP && para.stage == JobStage::POST_JOB) {
        InitRepoHandler();
        DeleteVolumes();
    }
    if (m_jobHandle->GetJobType() != JobType::RESTORE) {
        return SUCCESS;
    }
    if (para.stage == JobStage::EXECUTE_SUB_JOB) {
        m_reportParam = {
            "virtual_plugin_restore_job_execute_subjob_start_label",
            JobLogLevel::TASK_LOG_INFO,
            SubJobStatus::RUNNING, 0, 0 };
    }
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::GetMachineMetadata(VMInfo &vmInfo)
{
    if (!DoGetMachineMetadata()) {
        ERRLOG("Get machine metadata failed, %s", m_taskId.c_str());
        return FAILED;
    }
    vmInfo = m_vmInfo;
    return SUCCESS;
}

bool ApsaraStackProtectEngine::DoGetMachineMetadata()
{
    int32_t ret = SUCCESS;
    TP_START("TP_GetMachineMetaData", 1, &ret);
    TP_END
    if (ret == FAILED) {
        return false;
    }
    if (m_machineMetaCached) {
        DBGLOG("Machine metadata cached, %s", m_taskId.c_str());
        return true;
    }

    if (!GetInstanceDetail(m_jobHandle->GetApp().id, m_instance)) {
        ERRLOG("Get instance detail failed.");
        return false;
    }
    if (TransInstanceDetail2VMInfo(m_instance, m_vmInfo) != SUCCESS) {
        ERRLOG("Translate ServerDetail to VMInfo failed, %s", m_taskId.c_str());
        return false;
    }
    
    m_machineMetaCached = true;
    DBGLOG("Do get machine metadata success, %s", m_taskId.c_str());
    return true;
}

int32_t ApsaraStackProtectEngine::GetDisksOfInstance(const std::string &instanceId, DiskArray &diskList)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ss", m_regionId.c_str(), instanceId.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_disks", pArgs,
        response) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return FAILED;
    }
    DiskResponse resourceResponse;
    if (Utils::TransStrToApsaraApiResponse(response, resourceResponse) != SUCCESS) {
        ERRLOG("Failed to parse list disks function result.");
        return FAILED;
    }
    diskList = resourceResponse.m_disks;
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::TransInstanceDetail2VMInfo(const Instance &ins, VMInfo &vmInfo)
{
    vmInfo.m_moRef = ins.m_id;
    vmInfo.m_uuid = ins.m_id;
    vmInfo.m_name = ins.m_name;
    vmInfo.m_location = ins.m_regionId;
    Instance tmpIns = ins;
    if (!Module::JsonHelper::StructToJsonString(tmpIns, vmInfo.m_extendInfo)) {
        ERRLOG("Failed to convert instance detail.");
        return FAILED;
    }
    DiskArray diskList;
    if (GetDisksOfInstance(ins.m_id, diskList) != SUCCESS) {
        ERRLOG("Get disks of instance %s failed.", ins.m_id.c_str());
        return FAILED;
    }

    for (const auto &disk : diskList.m_disk) {
        VolInfo volInfo;
        if (disk.m_category == "san_efficiency" || disk.m_category == "san_ssd") {
            continue;   // 过滤共享云盘
        }
        TransDisk2VolInfo(disk, volInfo);
        volInfo.m_vmMoRef = ins.m_id;
        vmInfo.m_volList.push_back(volInfo);
    }
    DBGLOG("Trans ServerDetail to VMInfo success, %s", m_taskId.c_str());
    return SUCCESS;
}

void ApsaraStackProtectEngine::TransDisk2VolInfo(const Disk &disk, VolInfo &volInfo)
{
    volInfo.m_moRef = disk.m_id;
    volInfo.m_uuid = disk.m_id;
    volInfo.m_name = disk.m_name;
    volInfo.m_type = disk.m_category;
    volInfo.m_volSizeInBytes = disk.m_size * GB_SIZE;
    volInfo.m_bootable = disk.m_type;
    volInfo.m_volumeType = disk.m_category;
    volInfo.m_location = disk.m_regionId;
    volInfo.m_extendInfo = disk.m_zoneId;
    Disk volObj = disk;
    Module::JsonHelper::StructToJsonString(volObj, volInfo.m_metadata);
    volInfo.m_attachPoints.push_back(VolAttachMents(disk.m_device));
    
    DBGLOG("Trans VolumeDetail to VolInfo success.%s %s %s Volume size: %llu  volume type: %s. %s.",
        volInfo.m_uuid.c_str(), volInfo.m_name.c_str(), volInfo.m_type.c_str(), volInfo.m_volSizeInBytes,
        volInfo.m_volumeType.c_str(), m_taskId.c_str());
    return;
}

PyObject* ApsaraStackProtectEngine::SetCommonPara()
{
    PyObject* pArgs = nullptr;
    if (!Module::JsonHelper::JsonStringToStruct(m_appEnv.extendInfo, m_envParam)) {
        ERRLOG("Failed to parse resource groups function result.");
        return pArgs;
    }
    CertInfo cert;
    if (!Module::JsonHelper::JsonStringToStruct(m_appEnv.auth.extendInfo, cert)) {
        ERRLOG("Failed to parse resource groups function result.");
        return pArgs;
    }
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = pArgs = Py_BuildValue(COMMON_FORMAT.c_str(), m_appEnv.endpoint.c_str(), COMMON_RESOURCEGROUP.c_str(),
            m_envParam.m_organizationId.c_str(), m_regionId.c_str(), m_appEnv.auth.authkey.c_str(),
            m_appEnv.auth.authPwd.c_str(), DEFAULT_TIMEOUT, cert.m_certification.c_str(), cert.m_revocationList.c_str(),
            m_envParam.m_proxy.c_str());
    }
    std::string erroInfo;
    if (pArgs == nullptr) {
        ERRLOG("Py build value Err!");
        ExecutePython::GetInstance()->PrintErr(erroInfo);
        return pArgs;
    }
    return pArgs;
}

int32_t ApsaraStackProtectEngine::GetVolumesMetadata(const VMInfo &vmInfo,
    std::unordered_map<std::string, std::string> &volsMetadata)
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::GetVolumeHandler(const VolInfo &volInfo, std::shared_ptr<VolumeHandler> &volHandler)
{
    if (!InitJobPara()) {
        ERRLOG("Init Job Para failed.");
        return FAILED;
    }
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    std::shared_ptr<ApsaraVolumeHandler> apsaraStackVolHandler =
        std::make_shared<ApsaraVolumeHandler>(GetJobHandle(), volInfo, m_jobId, m_subJobId);
    if (apsaraStackVolHandler->InitializeVolumeInfo(APSARA_STACK_CONF) != SUCCESS) {
        ERRLOG("Initialize volume info failed.");
        return FAILED;
    }
    volHandler = apsaraStackVolHandler;
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::DetachVolumeHandle(const VolInfo &volObj, const std::string& serverId)
{
    std::string attachInstanceId;
    if (!GetServerWhichAttachedVolume(volObj, attachInstanceId)) {
        ERRLOG("Get disk(%s) detail failed. %s.", volObj.m_uuid.c_str(), m_taskId.c_str());
        return FAILED;
    }
    if (attachInstanceId == "") {
        INFOLOG("Disk(%s) have been detached. %s.", volObj.m_uuid.c_str(), m_taskId.c_str());
        return SUCCESS;
    }
    for (int i = 0; i < COMMON_RETRY_TIME; ++i) {
        if (RequestDetachVolume(serverId, volObj.m_uuid)) {
            INFOLOG("Detach disk(%s) success. %s.", volObj.m_uuid.c_str(), m_taskId.c_str());
            return SUCCESS;
        }
        WARNLOG("Request detach disk failed. %s.", m_taskId.c_str());
    }
    ERRLOG("DetachVolumeHandle failed. %s.", m_taskId.c_str());
    return FAILED;
}

bool ApsaraStackProtectEngine::RequestAttachVolume(const std::string& instanceId, const std::string& diskId,
    const std::string& device, const VolInfo &volObj)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        std::string diskKey = "";
        if (volObj.m_bootable == "system") {
            for (int index = 0; index < m_restorePara->restoreSubObjects.size(); ++index) {
                Json::Value targetVolume;
                if (InitParaAndGetTargetVolume(m_restorePara->restoreSubObjects[index], targetVolume) != SUCCESS) {
                    ERRLOG("InitParaAndGetTargetVolume failed, %s", m_taskId.c_str());
                    return false;
                }
                if (targetVolume.isMember("password")) {
                    diskKey = targetVolume["password"].asString();
                    break;
                }
            }
        }
        if (!diskKey.empty()) {
            INFOLOG("Start to attach system disk %s to host %s.", diskId.c_str(), instanceId.c_str());
            pArgs = Py_BuildValue("sssss", instanceId.c_str(), diskId.c_str(), "False", "True",
                diskKey.c_str());
        } else {
            pArgs = Py_BuildValue("ss", instanceId.c_str(), diskId.c_str());
        }
        diskKey = "";
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "attach_disk", pArgs,
        rspStr) != SUCCESS) {
        ERRLOG("Call python funtion attach_disk failed.");
        return false;
    }
    std::vector<std::string> intermediateState = {"Attaching"};
    VolInfo volInfo;
    volInfo.m_location = m_regionId;
    std::shared_ptr<VolumeHandler> volHandler;
    GetVolumeHandler(volInfo, volHandler);
    if (volHandler == nullptr) {
        ERRLOG("Null ptr of volume handler.");
        return false;
    }
    if (volHandler->DoWaitVolumeStatus(diskId, APSARA_DISK_IN_USE_STATUS, intermediateState) != SUCCESS) {
        ERRLOG("Wait disk(%s) status in-use timeout.", diskId.c_str());
        return false;
    }
    INFOLOG("Attach disk(%s) to instance %s success.", diskId.c_str(), instanceId.c_str());
    return true;
}

int32_t ApsaraStackProtectEngine::GetDiskDetail(const std::string &diskId, Disk &diskDetail,
    int32_t &erroCode)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ssiis", m_regionId.c_str(), "", 1, 10, diskId.c_str());
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_disks", pArgs,
        rspStr) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return FAILED;
    }
    DiskResponse response;
    if (Utils::TransStrToApsaraApiResponse(rspStr, response) != SUCCESS) {
        ERRLOG("Failed to parse get disk info function result.");
        return FAILED;
    }
    if (response.m_total == 0) {
        erroCode = Module::SC_NOT_FOUND;
        ERRLOG("Disk %s dont find.", diskId.c_str());
        return FAILED;
    }
    diskDetail = response.m_disks.m_disk[0];
    INFOLOG("Get disk %s info success.", diskId.c_str());
    return SUCCESS;
}

bool ApsaraStackProtectEngine::RequestDetachVolume(const std::string& serverId, const std::string& volumeId)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ss", serverId.c_str(), volumeId.c_str());
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "detach_disk", pArgs,
        rspStr) != SUCCESS) {
        ERRLOG("Call python funtion detach_disk failed.");
        return false;
    }
    VolInfo volInfo;
    volInfo.m_location = m_regionId;
    std::shared_ptr<VolumeHandler> volHandler;
    GetVolumeHandler(volInfo, volHandler);
    if (volHandler == nullptr) {
        ERRLOG("Null ptr of volume handler.");
        return false;
    }
    if (volHandler->DoWaitVolumeStatus(volumeId, APSARA_DISK_AVAILABLE_STATUS) != SUCCESS) {
        ERRLOG("Wait disk(%s) status available timeout.", volumeId.c_str());
        return false;
    }
    return true;
}

bool ApsaraStackProtectEngine::GetServerWhichAttachedVolume(const VolInfo &volInfo, std::string& serverId)
{
    Disk disk;
    serverId = "";
    int32_t erro;
    if (GetDiskDetail(volInfo.m_uuid, disk, erro) != SUCCESS) {
        ERRLOG("Get disk(%s) detail failed.", volInfo.m_uuid.c_str());
        return false;
    }

    if (disk.m_attachMents.m_attachment.size() < 1) {
        INFOLOG("Disk(%s) no attach to any server.", volInfo.m_uuid.c_str());
        return true;
    }
    serverId = disk.m_attachMents.m_attachment[0].m_instanceId;
    INFOLOG("vol(%s) been attached to server(%s).", volInfo.m_uuid.c_str(), serverId.c_str());
    return true;
}

void ApsaraStackProtectEngine::DeleteVolumesAction(const std::string &file, NewCreatedVolumeList &deleteFailVolumes)
{
    if (m_metaRepoHandler == nullptr) {
        ERRLOG("m_metaRepoHandler is nullptr, RepositoryHandler init failed.");
        return;
    }
    if (!m_metaRepoHandler->Exists(file)) {
        INFOLOG("New volume file(%s) does not exist, %s", file.c_str(), m_taskId.c_str());
        return;
    }
    INFOLOG("Begin to delele volume from file(%s).", file.c_str());
    uint32_t flag = Module::ConfigReader::getInt("ApsaraStackConfig", "DeleteSnapshotVolume");
    if (flag == NOT_DELETE_SNAPSHOT_VOLUME) {
        WARNLOG("Save volume flag exist.");
        return;
    }
    NewCreatedVolumeList volumeList;
    if (Utils::LoadFileToStructWithRetry(m_metaRepoHandler, file, volumeList) != SUCCESS) {
        ERRLOG("Load volume file(%s) failed, %s", file.c_str(), m_taskId.c_str());
        return;
    }
    VolInfo volObj;
    Disk diskDetail;
    Volume vol;
    for (const auto &volume : volumeList.m_volumelist) {
        volObj.m_uuid = volume.m_id;
        INFOLOG("Begin to delete disk(%s).", volObj.m_uuid.c_str());
        int32_t erro = 0;
        if (GetDiskDetail(volObj.m_uuid, diskDetail, erro) != SUCCESS && erro != Module::SC_NOT_FOUND) {
            ERRLOG("Show disk(%s) detail failed, %s", volObj.m_uuid.c_str(), m_taskId.c_str());
            deleteFailVolumes.m_volumelist.push_back(volume);
            continue;
        }
        if (erro == Module::SC_NOT_FOUND) {
            INFOLOG("Disk %s has been deleted.", volObj.m_uuid.c_str());
            continue;
        }
        // 若卷处于挂载状态，需要先卸载卷
        if (diskDetail.m_status == APSARA_DISK_IN_USE_STATUS &&
            DetachVolumeHandle(volObj, diskDetail.m_attachMents.m_attachment[0].m_instanceId) != SUCCESS) {
            deleteFailVolumes.m_volumelist.push_back(volume);
            continue;
        }
        if (DeleteVolume(volObj) != SUCCESS) {
            ERRLOG("Delete volume(%s) failed, %s", volObj.m_uuid.c_str(), m_taskId.c_str());
            deleteFailVolumes.m_volumelist.push_back(volume);
            continue;
        }
        INFOLOG("Delete volume(%s) success.", volObj.m_uuid.c_str());
    }
    return;
}

int32_t ApsaraStackProtectEngine::DoDeleteVolume(const VolInfo &volObj)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("s", volObj.m_uuid.c_str());
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "delete_disk",
        pArgs, rspStr) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return FAILED;
    }

    Disk diskInfo;
    int32_t erro;
    for (int i = 0; i < COMMON_RETRY_TIME; ++i) {
        if (GetDiskDetail(volObj.m_uuid, diskInfo, erro) != SUCCESS && erro == Module::SC_NOT_FOUND) {
            INFOLOG("Disk %s delete success.", volObj.m_uuid.c_str());
            return SUCCESS;
        }
        sleep(COMMON_WAIT_TIME);
        WARNLOG("Delete disk %s failed, disk status: %s.", volObj.m_uuid.c_str(), diskInfo.m_status.c_str());
    }
    ERRLOG("Delete disk %s failed.", volObj.m_uuid.c_str());
    return FAILED;
}

bool ApsaraStackProtectEngine::DeleteSnapshotCreateVolumes(const VolSnapInfo &snapInfo)
{
    INFOLOG("Begin to get snapshot(%s) clone volumes.", snapInfo.m_snapshotId.c_str());
    InitRepoHandler();
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ssiiss", m_regionId.c_str(), "", 1, 100, "", snapInfo.m_snapshotId.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_disks", pArgs,
        response) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return false;
    }
    DiskResponse resourceResponse;
    if (Utils::TransStrToApsaraApiResponse(response, resourceResponse) != SUCCESS) {
        ERRLOG("Failed to parse list disks function result.");
        return false;
    }
    DiskArray diskList = resourceResponse.m_disks;
    return FilterAndDeleteDisks(diskList);
}

bool ApsaraStackProtectEngine::FilterAndDeleteDisks(const DiskArray &diskList)
{
    std::string historyVolFile = m_metaRepoPath + VIRT_PLUGIN_VOLUME_TOBEDELETED;
    NewCreatedVolumeList deleteFailVols;
    if (Utils::LoadFileToStructWithRetry(m_metaRepoHandler, historyVolFile, deleteFailVols) != SUCCESS) {
        WARNLOG("Load volume file(%s) failed, %s", historyVolFile.c_str(), m_taskId.c_str());
    }
    VolInfo volObj;
    Volume volume;
    bool deleteSuccess = true;
    for (const auto &disk : diskList.m_disk) {
        if (!MatchVolumeName(disk.m_name)) {
            WARNLOG("Disk(%s) does not match.", disk.m_name.c_str());
            continue;
        }
        if (CheckIfPreHookDelete(disk.m_id, deleteFailVols)) {
            deleteSuccess = false;
            continue;
        }
        volObj.m_uuid = disk.m_id;
        Utils::ConvertDiskToVolume(disk, volume);
        if (disk.m_status == APSARA_DISK_IN_USE_STATUS &&
            DetachVolumeHandle(volObj, disk.m_attachMents.m_attachment[0].m_instanceId) != SUCCESS) {
            deleteFailVols.m_volumelist.push_back(volume); // 更新删除失败的卷列表
            deleteSuccess = false;
            continue;
        }
        if (DeleteVolume(volObj) != SUCCESS) {
            ERRLOG("Delete volume(%s) failed, %s", volObj.m_uuid.c_str(), m_taskId.c_str());
            deleteFailVols.m_volumelist.push_back(volume); // 更新删除失败的卷列表
            deleteSuccess = false;
            continue;
        }
        DBGLOG("Delete volume(%s) success.", volObj.m_uuid.c_str());
    }
    SaveDeleteFailVolumes(historyVolFile, deleteFailVols);
    return deleteSuccess;
}

bool ApsaraStackProtectEngine::CheckVolumesIfDelete(const VolResidualInfo &volResidualInfo)
{
    Disk disk;
    for (const auto &residualVol : volResidualInfo.m_volumelist) {
        int32_t erro;
        if (GetDiskDetail(residualVol.m_id, disk, erro) == SUCCESS) {
            INFOLOG("Show clone disk(%s) detail success, %s", residualVol.m_id.c_str(), m_taskId.c_str());
            return false;
        }
    }
    return true;
}

int32_t ApsaraStackProtectEngine::ReplaceVolume(const VolInfo &volObj)
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::CreateMachine(VMInfo &vmInfo)
{
    vmInfo.m_uuid = m_jobHandle->GetApp().id;
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::GenVolPair(VMInfo &vmObj, const VolInfo &copyVol,
    const ApplicationResource &targetVol, VolMatchPairInfo &volPairs)
{
    Json::Value targetVolume;
    if (InitParaAndGetTargetVolume(targetVol, targetVolume) != SUCCESS) {
        ERRLOG("InitParaAndGetTargetVolume failed, %s", m_taskId.c_str());
        return FAILED;
    }
    if (!InitRestoreLevel()) {
        INFOLOG("Init RestoreLevel failed.");
        return FAILED;
    }
    std::string targetUuid = targetVolume["id"].asString();
    std::string name = copyVol.m_name;
    std::string isNewDisk = targetVolume["isNewDisk"].asString();
    VolInfo dstVolInfo;
    if (isNewDisk == "true") {
        VolInfo backupvolInfo;
        if (LoadCopyVolumeMatadata(copyVol.m_uuid, backupvolInfo) == FAILED) {
            ERRLOG("LoadCopyVolumeMatadata failed.");
            return FAILED;
        }
        Disk diskDetail;
        if (CreateVolume(backupvolInfo, "", "", backupvolInfo.m_datastore, dstVolInfo) == FAILED) {
            ERRLOG("Create volume failed.");
            return FAILED;
        }
        INFOLOG("Creat disk %s success.", dstVolInfo.m_uuid.c_str());
    } else {
        if (!GetVolumeInfo(targetUuid, dstVolInfo)) {
            ERRLOG("Get volume(%s) failed, %s", targetUuid.c_str(), m_taskId.c_str());
            return FAILED;
        }
        if (m_restoreLevel == RestoreLevel::RESTORE_TYPE_VM && dstVolInfo.m_bootable == "system") {
            PowerOffMachine(vmObj);
        }
    }
    VolPair restoreVolPair;
    restoreVolPair.m_originVol = copyVol;
    restoreVolPair.m_targetVol = dstVolInfo;
    volPairs.m_volPairList.push_back(restoreVolPair);
    INFOLOG("Copy vol(%s, %s), target vol(%s, %s), %s", copyVol.m_name.c_str(), copyVol.m_uuid.c_str(),
        dstVolInfo.m_name.c_str(), dstVolInfo.m_uuid.c_str(), m_taskId.c_str());
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::DeleteMachine(const VMInfo &vmInfo)
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::RenameMachine(const VMInfo &vmInfo, const std::string &newName)
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::PowerOnMachineHandle(const VMInfo &vmInfo)
{
    // check if server can power on
    Instance instance;
    if (!GetInstanceDetail(vmInfo.m_uuid, instance)) {
        ERRLOG("Get server details failed. %s", m_taskId.c_str());
        return FAILED;
    }
    std::string& status = instance.m_status;
    if (status == APSARA_STATUS_RUNNING) {
        INFOLOG("Machine already power on.");
        return SUCCESS;
    }
    if (status != APSARA_STATUS_STOPPED) {
        ERRLOG("Machine status err.");
        return FAILED;
    }
    // power on
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("s", vmInfo.m_uuid.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "start_instance",
        pArgs, response) != SUCCESS) {
        ERRLOG("Call python funtion start_instance failed.");
        return FAILED;
    }
    InstanceResponse instanceResponse;
    if (Utils::TransStrToApsaraApiResponse(response, instanceResponse) != SUCCESS) {
        ERRLOG("Failed to parse list disks function result.");
        return FAILED;
    }
    // check if server success active
    if (!GetAndCheckMachineStatus(vmInfo, APSARA_STATUS_RUNNING, m_commonWaitTimes)) {
        ERRLOG("Power on machine failed. %s", m_taskId.c_str());
        return FAILED;
    }
    INFOLOG("Power on machine success.");
    return SUCCESS;
}

bool ApsaraStackProtectEngine::GetAndCheckMachineStatus(const VMInfo &vmInfo, const std::string& status, int waitTimes)
{
    Instance instance;
    for (int32_t retryTimes = 0; retryTimes <= waitTimes; retryTimes++) {
        if (!GetInstanceDetail(vmInfo.m_uuid, instance)) {
            ERRLOG("Get machine(%s) id(%s) status(%s) fail, %s", vmInfo.m_name.c_str(), vmInfo.m_uuid.c_str(),
                status.c_str(), m_taskId.c_str());
            return false;
        }
        if (instance.m_status == status) {
            INFOLOG("Check machine(%s) id(%s) status(%s) success, %s", vmInfo.m_name.c_str(), vmInfo.m_uuid.c_str(),
                status.c_str(), m_taskId.c_str());
            return true;
        }
        sleep(m_commonWaitInterval);
    }
    ERRLOG("Check machine(%s) id(%s) status(%s) fail, %s", vmInfo.m_name.c_str(), vmInfo.m_uuid.c_str(),
        status.c_str(), m_taskId.c_str());
    return false;
}

int ApsaraStackProtectEngine::PostHook(const ExecHookParam &para)
{
    if (!InitRestoreLevel()) {
        return FAILED;
    }
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::PowerOffMachine(const VMInfo &vmInfo)
{
    int ret = PowerOffMachineHandle(vmInfo);
    m_reportArgs = { vmInfo.m_uuid.c_str() };
    if (ret != SUCCESS) {
        m_reportParam = {
        "virtual_plugin_restore_job_power_off_vm_failed_label",
        JobLogLevel::TASK_LOG_ERROR,
        SubJobStatus::FAILED, 0, 0 };
        return ret;
    }
    m_reportParam = {
        "virtual_plugin_restore_job_power_off_vm_success_label",
        JobLogLevel::TASK_LOG_INFO,
        SubJobStatus::RUNNING, 0, 0 };
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::PowerOffMachineHandle(const VMInfo &vmInfo)
{
    // check if server can power off
    Instance instance;
    if (!GetInstanceDetail(vmInfo.m_uuid, instance)) {
        ERRLOG("Get server details failed. %s", m_taskId.c_str());
        return FAILED;
    }
    std::string& status = instance.m_status;
    if (status == APSARA_STATUS_STOPPED) {
        INFOLOG("Machine already power off.");
        return SUCCESS;
    }
    if (status != APSARA_STATUS_RUNNING) {
        ERRLOG("machine status: %s can not power off.", status.c_str());
        return FAILED;
    }

    // power off
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("s", vmInfo.m_uuid.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "stop_instance",
        pArgs, response) != SUCCESS) {
        ERRLOG("Call python funtion stop_instance failed.");
        return FAILED;
    }
    // check if server success shutoff
    if (!GetAndCheckMachineStatus(vmInfo, APSARA_STATUS_STOPPED, m_commonWaitTimes)) {
        ERRLOG("Power off machine failed. %s", m_taskId.c_str());
        return FAILED;
    }
    INFOLOG("Power off machine(%s, %s) success.", vmInfo.m_name.c_str(), vmInfo.m_uuid.c_str());
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::AllowBackupInLocalNode(const AppProtect::BackupJob& job, int32_t &errorCode)
{
    int32_t ret = CheckEnvConnection(job.protectEnv, job.protectObject.id, errorCode);
    TP_START("TP_AllowBackupInLocalNode", 1, &ret);
    TP_END
    return ret;
}

int32_t ApsaraStackProtectEngine::AllowBackupSubJobInLocalNode(const AppProtect::BackupJob &job,
    const AppProtect::SubJob &subJob, int32_t &errorCode)
{
    return CheckEnvConnection(job.protectEnv, job.protectObject.id, errorCode);
}

int32_t ApsaraStackProtectEngine::AllowRestoreInLocalNode(const AppProtect::RestoreJob& job, int32_t &errorCode)
{
    return CheckEnvConnection(job.targetEnv, job.targetObject.id, errorCode);
}

int32_t ApsaraStackProtectEngine::AllowRestoreSubJobInLocalNode(const AppProtect::RestoreJob& job,
    const AppProtect::SubJob& subJob, int32_t &errorCode)
{
    return CheckEnvConnection(job.targetEnv, job.targetObject.id, errorCode);
}

int32_t ApsaraStackProtectEngine::CheckBeforeBackup()
{
    if (!CheckVMStatus()) {
        ERRLOG("Check VM Status Failed.");
        return FAILED;
    }
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::CheckBeforeCreateSnapshot(const std::vector<VolInfo> &volList)
{
    return SUCCESS;
}

bool ApsaraStackProtectEngine::CheckVMStatus()
{
    std::string jobTypeStatus;
    if (m_jobHandle->GetJobType() == JobType::BACKUP) {
        jobTypeStatus = "SupportBackupStatus";
        m_appEnv = m_backupPara->protectEnv;
    } else if (m_jobHandle->GetJobType() == JobType::RESTORE) {
        jobTypeStatus = "SupportRestoreStatus";
        m_appEnv = m_restorePara->targetEnv;
    } else {
        ERRLOG("check vm status failed, jobtype: %s. %s", m_jobHandle->GetJobType(), m_taskId.c_str());
        return false;
    }
    std::string supportStatus = Module::ConfigReader::getString("ApsaraStackConfig", jobTypeStatus);
    Instance ins;
    if (!GetInstanceDetail(m_jobHandle->GetApp().id, ins)) {
        ERRLOG("Get instance detail failed.");
        return false;
    }
    if (supportStatus.find(ins.m_status) == std::string::npos) {
        ERRLOG("Instance status %s dont support backup.", ins.m_status.c_str());
        return false;
    }
    DBGLOG("Check instance status success.");
    return true;
}

bool ApsaraStackProtectEngine::CheckTargetVolume(const VolInfo &copyVolObj, const ApplicationResource &restoreVol)
{
    Json::Value volExtendInfo;
    if (!Module::JsonHelper::JsonStringToJsonValue(restoreVol.extendInfo, volExtendInfo)) {
        ERRLOG("Convert %s failed, %s", WIPE_SENSITIVE(restoreVol.extendInfo.c_str()), m_taskId.c_str());
        return false;
    }
    std::string targetVolsInfoStr = volExtendInfo["targetVolume"].asString();
    DBGLOG("TargetVolsInfo %s. %s", targetVolsInfoStr.c_str(), m_taskId.c_str());
    Json::Value targetVolume;
    if (!Module::JsonHelper::JsonStringToJsonValue(targetVolsInfoStr, targetVolume)) {
        ERRLOG("Transfer %s failed, %s.", targetVolsInfoStr.c_str(), m_taskId.c_str());
        return false;
    }
    // 需要新创的卷不需要检查，直接返回
    std::string isNewDisk = targetVolume["isNewDisk"].asString();
    if (isNewDisk == "true") {
        INFOLOG("New volume dont check.");
        return true;
    }
    std::string diskId = targetVolume["id"].asString();
    Disk diskInfo;
    int32_t erro;
    if (GetDiskDetail(diskId, diskInfo, erro) != SUCCESS) {
        ERRLOG("Get disk %s detail failed.", diskId.c_str());
        return false;
    }
    if (!CheckVolStatus(diskInfo)) {
        return false;
    }
    if (!CheckVolMode(copyVolObj, diskInfo)) {
        return false;
    }
    uint64_t sizeInBytes = diskInfo.m_size * GB_SIZE;
    if (sizeInBytes < copyVolObj.m_volSizeInBytes) {
        ERRLOG("Copy vol size %llu, target vol size %llu, %s", copyVolObj.m_volSizeInBytes, sizeInBytes,
            m_taskId.c_str());
        return false;
    }
    DBGLOG("Target volume(%s) check success, %s", targetVolume["uuid"].asString().c_str(), m_taskId.c_str());
    return true;
}

/**
 * @brief 检查卷状态
 *
 * @param diskInfo
 * @return true
 * @return false
 */
bool ApsaraStackProtectEngine::CheckVolStatus(const Disk &diskInfo)
{
    std::string supportStatus = Module::ConfigReader::getString("ApsaraStackConfig", "VolSupportStatus");
    INFOLOG("supportStatus %s", supportStatus.c_str());
    std::set<std::string> statusList;
    boost::split(statusList, supportStatus, boost::is_any_of(","));
    std::set<std::string>::iterator pos = statusList.find(diskInfo.m_status);
    if (pos == statusList.end()) {
        ERRLOG("Can not exec restore for vol status: %s, %s", diskInfo.m_status.c_str(), m_taskId.c_str());
        return false;
    }
    return true;
}

/**
 * @brief 盘类型检查(启动盘与非启动盘数据不能互相恢复)
 *
 * @return true
 * @return false
 */
bool ApsaraStackProtectEngine::CheckVolMode(const VolInfo &copyDisk, const Disk &targetDisk)
{
    if (!targetDisk.m_portable) {
        ERRLOG("Disk not portable, cannot restore: %s, %s", targetDisk.m_name.c_str(), m_taskId.c_str());
        return false;
    }
    if (targetDisk.m_type == "data") {
        DBGLOG("No need to check for type(%s) volume. %s", targetDisk.m_type.c_str(), m_taskId.c_str());
        return true;
    }
    if (copyDisk.m_bootable != targetDisk.m_type) {
        ERRLOG("Copy disk type is %s, target type is %s, %s", copyDisk.m_bootable.c_str(),
            targetDisk.m_type.c_str(), m_taskId.c_str());
        return false;
    }
    DBGLOG("Disk type check success. %s", m_taskId.c_str());
    return true;
}

bool ApsaraStackProtectEngine::GetInstanceDetail(const std::string &instanceId, Instance& ins)
{
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("siis", m_regionId.c_str(), DEFAULT_PAGE_NUMBER, DEFAULT_PAGE_SIZE,
            instanceId.c_str());
    }
    std::string response;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "list_instances", pArgs,
        response) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return false;
    }
    InstanceResponse instances;
    if (Utils::TransStrToApsaraApiResponse(response, instances) != SUCCESS) {
        ERRLOG("Failed to parse get instance function result.");
        return false;
    }
    if (instances.m_total != 1) {
        ERRLOG("Scan %d instances detail.", instances.m_total);
        return false;
    }
    ins = instances.m_instances.m_instance[instances.m_total - 1];
    INFOLOG("Get detail of instance %s success.", instanceId.c_str());
    return true;
}

int32_t ApsaraStackProtectEngine::CheckEnvConnection(
    const AppProtect::ApplicationEnvironment &env, const std::string &vmId, int32_t &errorCode)
{
    m_appEnv = env;
    m_pythonObject = env.name;
    PyObject* pArgs = nullptr;
    std::string response;
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "connect_check", pArgs,
        response) != SUCCESS) {
        ERRLOG("Failed to CallObjFunction connect_check.");
        return FAILED;
    }
    APSResponse checkBody;
    if (!Module::JsonHelper::JsonStringToStruct(response, checkBody)) {
        ERRLOG("Failed to parse connect check function result.");
        return FAILED;
    }
    if (checkBody.m_statusCode != Module::SC_OK) {
        ERRLOG("Check env connection failed.");
        return FAILED;
    }
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::CheckBackupJobType(const VirtPlugin::JobTypeParam& jobTypeParam, bool& checkRet)
{
    m_appEnv = jobTypeParam.m_job.protectEnv;
    ExecutePython::GetInstance()->SetClassPara(SetCommonPara());
    std::vector<std::string> diskIdList;
    if (jobTypeParam.m_job.protectSubObject.size() == 0) {
        /* backup all volumes */
        DiskArray diskList;
        if (GetDisksOfInstance(jobTypeParam.m_job.protectObject.id, diskList) != SUCCESS) {
            ERRLOG("Get disks of instance %s failed.", jobTypeParam.m_job.protectObject.id.c_str());
            return FAILED;
        }
        for (const auto &disk : diskList.m_disk) {
            if (disk.m_category == "san_efficiency" || disk.m_category == "san_ssd") {
                continue;   // 过滤共享云盘
            }
            diskIdList.push_back(disk.m_id);
        }
    } else {
        /* backup target volumes */
        for (const auto &disk : jobTypeParam.m_job.protectSubObject) {
            diskIdList.push_back(disk.id);
        }
    }
    std::string metaRepoPath;
    std::shared_ptr<RepositoryHandler> metaRepoHandler = nullptr;
    if (Utils::GetMetaRepoPath(metaRepoPath, metaRepoHandler, jobTypeParam.m_job) != SUCCESS) {
        ERRLOG("GetMetaRepoPath failed.");
        return FAILED;
    }
    for (const auto &id : diskIdList) {
        std::string volHashFile = metaRepoPath + VIRT_PLUGIN_VOLUMES_HASH_DIR + id + "_hash.record";
        if (!metaRepoHandler->Exists(volHashFile)) {
            ERRLOG("Disk %s hash file dont exit.", id.c_str());
            checkRet = false;
            return SUCCESS;
        }
    }
    checkRet = true;
    return SUCCESS;
}

void ApsaraStackProtectEngine::ListApplicationResource(std::vector<ApplicationResource>& returnValue,
    const ApplicationEnvironment& appEnv, const Application& application, const ApplicationResource& parentResource)
{
    return;
}

void ApsaraStackProtectEngine::DiscoverHostCluster(ApplicationEnvironment& returnEnv,
    const ApplicationEnvironment& appEnv)
{
    return;
}

int32_t ApsaraStackProtectEngine::DoCreateVolume(const VolInfo &backupVol, const std::string &volMetaData,
    const std::string &vmMoRef, const DatastoreInfo &dsInfo, VolInfo &newVol)
{
    int32_t diskSize = backupVol.m_volSizeInBytes / GB_SIZE;
    PyObject* pArgs;
    {
        std::lock_guard<std::mutex> guard(ExecutePython::m_pythonMutex);
        PyAllowThreads thread;
        pArgs = Py_BuildValue("ssssiss", backupVol.m_location.c_str(), "",
        backupVol.m_extendInfo.c_str(), backupVol.m_name.c_str(), diskSize, backupVol.m_type.c_str(), "");
    }
    std::string rspStr;
    if (ExecutePython::GetInstance()->CallObjFunction(APSARASTACK_API_CLASS, "create_disk", pArgs,
        rspStr) != SUCCESS) {
        ERRLOG("Call python funtion failed.");
        return FAILED;
    }
    APSCreateDiskResponse response;
    if (Utils::TransStrToApsaraApiResponse(rspStr, response) != SUCCESS) {
        ERRLOG("Failed to parse create disk function result.");
        return FAILED;
    }
    Disk newDiskInfo;
    int32_t erro;
    if (GetDiskDetail(response.m_diskId, newDiskInfo, erro) != SUCCESS) {
        ERRLOG("Get disk(%s) detail failed.", response.m_diskId.c_str());
        return FAILED;
    }
    TransDisk2VolInfo(newDiskInfo, newVol);
    newVol.m_bootable = backupVol.m_bootable;
    if (m_restoreLevel == RestoreLevel::RESTORE_TYPE_VM) {
        for (const auto &it : backupVol.m_attachPoints) {
            DBGLOG("Volume attach device is %s, %s", it.m_device.c_str(), m_taskId.c_str());
            newVol.m_attachPoints.push_back(VolAttachMents(it.m_device));
        }
    }
    std::vector<std::string> intermediateState = {"creating"};
    VolInfo volInfo;
    volInfo.m_location = m_regionId;
    std::shared_ptr<VolumeHandler> volHandler;
    GetVolumeHandler(volInfo, volHandler);
    if (volHandler == nullptr) {
        ERRLOG("Null ptr of volume handler.");
        return FAILED;
    }
    if (volHandler->DoWaitVolumeStatus(newVol.m_uuid, APSARA_DISK_AVAILABLE_STATUS, intermediateState) == FAILED) {
        ERRLOG("Wait volume %s available failed.", newVol.m_uuid.c_str());
        return FAILED;
    }
    newVol.m_newCreate = true;
    return SUCCESS;
}

bool ApsaraStackProtectEngine::GetVolumeInfo(const std::string &volId, VolInfo &volInfo)
{
    Disk DiskInfo;
    int32_t erro;
    if (GetDiskDetail(volId, DiskInfo, erro) != SUCCESS) {
        ERRLOG("Get disk(%s) detail failed.", volInfo.m_uuid.c_str());
        return false;
    }
    TransDisk2VolInfo(DiskInfo, volInfo);
    DBGLOG("Get volume detail success, %s", m_taskId.c_str());
    return true;
}

bool ApsaraStackProtectEngine::GetifDeleteSnapshot(AppProtect::BackupJobType backuptype,
    AppProtect::JobResult::type res)
{
    return true;
}

bool ApsaraStackProtectEngine::IfDeleteLatestSnapShot()
{
    return true;
}

int32_t ApsaraStackProtectEngine::CheckBeforeMount()
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::CancelLiveMount(const VMInfo &liveVm)
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::CreateLiveMount(const VMInfo &copyVm, VMInfo &newVm)
{
    return SUCCESS;
}

int32_t ApsaraStackProtectEngine::ActiveSnapConsistency(const SnapshotInfo& snapshotInfo, int32_t &erroCode)
{
    return SUCCESS;
}
}
